package com.fwmagic.flink.projectcase.twophase;

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;

public class MyTwoPhaseCommitSink extends TwoPhaseCommitSinkFunction<String,String,Void> {

    public MyTwoPhaseCommitSink() {
        super(new KryoSerializer<>(String.class, new ExecutionConfig()), VoidSerializer.INSTANCE);
    }

    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {
        super.snapshotState(context);
        System.err.println("==========4.snapshotState==============");
    }

    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {
        super.initializeState(context);
        System.err.println("==========2.initializeState==============");

    }

    @Override
    protected void invoke(String transaction, String value, Context context) throws Exception {
        System.err.println("==========invoke==============>"+value);
    }

    @Override
    protected String beginTransaction() throws Exception {
        System.err.println("============1.beginTransaction==============");
        return null;
    }

    @Override
    protected void preCommit(String transaction) throws Exception {
        System.err.println("============3.preCommit==============");

    }

    @Override
    protected void commit(String transaction) {
        System.err.println("============5.commit==============");
    }

    @Override
    protected void abort(String transaction) {
        System.err.println("============abort==============");

    }
}
